package com.wq.concurrency.aqs;

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;

/**
 * @Author: wangqiang20995
 * @Date:2019/7/30
 * @Description:
 * @Resource:
 */
public class ReentrantLockAndConditionObjectDemo {

    private final ReentrantLock lock = new ReentrantLock();

    private final Condition empty = lock.newCondition();

    private final Condition full = lock.newCondition();

    private final int[] tables;

    private volatile int number;

    public ReentrantLockAndConditionObjectDemo(int size) {
        this.tables = new int[size];
        this.number = 0;

        for (int i = 0; i < size; i++) {
            this.tables[i] = -1;
        }
    }

    //阻塞式
    public boolean put(int value) {
        lock.lock();

        try {

            while (this.number == this.tables.length) {
                printThreadWithMessage("table full and wait until not full");
                this.full.await();
            }

            printThreadWithMessage("put value[" + this.number + "]-->" + value);
            this.tables[this.number++] = value;

            //wake up a thread waited for table value
            this.empty.signal();

            return true;
        } catch (Exception e) {
            System.out.println("Exception:" + e.getMessage());
            return false;
        } finally {
            lock.unlock();
        }
    }

    public int get() {
        lock.lock();

        try {
            while (this.number == 0) {
                printThreadWithMessage("table empty and wait until not empty");
                this.empty.await();
            }

            int value = this.tables[--this.number];
            this.tables[this.number] = -1;//init value
            printThreadWithMessage("get value-->" + value);

            //wake up a thread waited for a slot
            this.full.signal();

            return value;
        } catch (Exception e) {
            e.printStackTrace();
            System.out.println("get exception:" + e.getMessage());
            return -1;
        } finally {
            lock.unlock();
        }
    }

    private static void printThreadWithMessage(String message) {
        System.out.printf("[Thread:%s]:%s\n", Thread.currentThread().getName(), message);
    }


    public boolean empty() {
        return this.number == 0;
    }

    public static void main(String[] args) {
        ReentrantLockAndConditionObjectDemo rlco = new ReentrantLockAndConditionObjectDemo(3);


        ExecutorService producer = new ThreadPoolExecutor(5, 20, 60, TimeUnit.SECONDS, new ArrayBlockingQueue<>(10));

        ExecutorService consumer = new ThreadPoolExecutor(5, 20, 60, TimeUnit.SECONDS, new ArrayBlockingQueue<>(10));

        for (int i = 0; i < 10; i++) {
            int finalI = i;
            producer.execute(() -> {
                rlco.put(finalI);
            });
        }

        for (int i = 0; i < 10; i++) {
            consumer.execute(() -> {
                rlco.get();
            });
        }

        //下面这段代码不需要，因为执行到这里的时候，所有的task都已经提交到线程池中了
        //而执行shutdown方法，线程池不再接受新的task，然后会执行阻塞队列中的task
        //等全部执行完了，就可以关闭了
//        try {
//            Thread.sleep(10000);
//        } catch (InterruptedException e) {
//            e.printStackTrace();
//        }

        producer.shutdown();
        consumer.shutdown();
    }
}
